package com.flink.hbase;

import org.apache.flink.api.common.functions.MapFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;

/**
 * User 数据转换器
 * 对从 HBase 读取的用户数据进行转换和清洗
 */
public class UserMapper implements MapFunction<User, User> {

    private static final Logger LOG = LoggerFactory.getLogger(UserMapper.class);
    private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd");

    @Override
    public User map(User user) throws Exception {
        if (user == null) {
            return null;
        }

        try {
            // 数据清洗和转换
            User cleanedUser = new User();
            
            // 用户ID处理
            cleanedUser.setUserId(cleanUserId(user.getUserId()));
            
            // 姓名处理
            cleanedUser.setName(cleanName(user.getName()));
            
            // 年龄处理
            cleanedUser.setAge(cleanAge(user.getAge()));
            
            // 性别处理
            cleanedUser.setGender(cleanGender(user.getGender()));
            
            // 时间戳处理
            cleanedUser.setTs(user.getTs() != null ? user.getTs() : System.currentTimeMillis());
            
            // 分区日期处理
            cleanedUser.setDt(cleanPartitionDate(user.getDt()));
            
            return cleanedUser;
            
        } catch (Exception e) {
            LOG.error("用户数据转换失败: {}", user, e);
            throw e;
        }
    }

    /**
     * 清洗用户ID
     */
    private String cleanUserId(String userId) {
        if (userId == null || userId.trim().isEmpty()) {
            throw new IllegalArgumentException("用户ID不能为空");
        }
        
        // 去除前后空格，转换为小写
        String cleaned = userId.trim().toLowerCase();
        
        // 验证格式（可根据具体业务规则调整）
        if (!cleaned.matches("^[a-z0-9_-]{1,50}$")) {
            LOG.warn("用户ID格式可能不规范: {}", userId);
        }
        
        return cleaned;
    }

    /**
     * 清洗姓名
     */
    private String cleanName(String name) {
        if (name == null || name.trim().isEmpty()) {
            return "Unknown";
        }
        
        // 去除前后空格
        String cleaned = name.trim();
        
        // 长度限制
        if (cleaned.length() > 100) {
            cleaned = cleaned.substring(0, 100);
            LOG.warn("姓名长度超限，已截断: {}", name);
        }
        
        // 移除特殊字符（保留中文、英文、数字、常见符号）
        cleaned = cleaned.replaceAll("[^\\u4e00-\\u9fa5a-zA-Z0-9\\s._-]", "");
        
        return cleaned.isEmpty() ? "Unknown" : cleaned;
    }

    /**
     * 清洗年龄
     */
    private Integer cleanAge(Integer age) {
        if (age == null) {
            return null;
        }
        
        // 年龄范围验证
        if (age < 0 || age > 150) {
            LOG.warn("年龄数值异常: {}", age);
            return null;
        }
        
        return age;
    }

    /**
     * 清洗性别
     */
    private String cleanGender(String gender) {
        if (gender == null || gender.trim().isEmpty()) {
            return "Unknown";
        }
        
        String cleaned = gender.trim().toLowerCase();
        
        // 标准化性别值
        switch (cleaned) {
            case "m":
            case "male":
            case "男":
            case "1":
                return "Male";
            case "f":
            case "female":
            case "女":
            case "0":
                return "Female";
            default:
                LOG.debug("性别值标准化: {} -> Unknown", gender);
                return "Unknown";
        }
    }

    /**
     * 清洗分区日期
     */
    private String cleanPartitionDate(String dt) {
        if (dt == null || dt.trim().isEmpty()) {
            // 使用当前日期作为默认值
            return LocalDate.now().format(DATE_FORMATTER);
        }
        
        String cleaned = dt.trim();
        
        try {
            // 验证日期格式
            LocalDate.parse(cleaned, DATE_FORMATTER);
            return cleaned;
        } catch (Exception e) {
            LOG.warn("分区日期格式不正确: {}, 使用当前日期", dt);
            return LocalDate.now().format(DATE_FORMATTER);
        }
    }

    /**
     * 数据质量检查
     */
    public static boolean isValidUser(User user) {
        if (user == null) {
            return false;
        }
        
        // 必须字段检查
        if (user.getUserId() == null || user.getUserId().trim().isEmpty()) {
            return false;
        }
        
        // 可以添加更多的数据质量规则
        return true;
    }

    /**
     * 获取数据质量评分（0-100）
     */
    public static int getDataQualityScore(User user) {
        if (user == null) {
            return 0;
        }
        
        int score = 0;
        
        // 用户ID（必须，40分）
        if (user.getUserId() != null && !user.getUserId().trim().isEmpty()) {
            score += 40;
        }
        
        // 姓名（20分）
        if (user.getName() != null && !user.getName().trim().isEmpty() && !"Unknown".equals(user.getName())) {
            score += 20;
        }
        
        // 年龄（20分）
        if (user.getAge() != null && user.getAge() > 0 && user.getAge() <= 150) {
            score += 20;
        }
        
        // 性别（10分）
        if (user.getGender() != null && ("Male".equals(user.getGender()) || "Female".equals(user.getGender()))) {
            score += 10;
        }
        
        // 时间戳（10分）
        if (user.getTs() != null && user.getTs() > 0) {
            score += 10;
        }
        
        return score;
    }
} 